package edu.nepu.flink.api.splitStream;

import edu.nepu.flink.api.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Date 2024/2/28 22:21
 * @Created by chenshuaijun
 */
public class SimpleSplitStream {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<WaterSensor> elementSource = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 11),
                new WaterSensor("s3", 4L, 5)
        );

        // 使用一种简单的分流方式
        SingleOutputStreamOperator<WaterSensor> stream1 = elementSource.filter(v -> "s1".equals(v.getId()));
        SingleOutputStreamOperator<WaterSensor> stream2 = elementSource.filter(v -> "s2".equals(v.getId()));
        SingleOutputStreamOperator<WaterSensor> stream3 = elementSource.filter(v -> "s3".equals(v.getId()));

        // 对于同一条流的数据我们使用filter过滤出我们想要的数据，通过这种方式来进行分流
        stream1.print("stream1");
        stream2.print("stream2");
        stream3.print("stream3");



        env.execute();
    }
}
